package priv.abadstring.study.rocketmq.javaapi;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * Consumer
 *
 * @author aBadString
 * @date 2021-08-01 16:55:26
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("haoke");
        consumer.setNamesrvAddr("abadstring.me:9876");

        // 订阅消息
        // 接收所有 tag 的消息
        // consumer.subscribe("song-song", "*");
        // 自定义 tag. a 或者 b
        consumer.subscribe("song-song", "a || b");

        // 收到消息的回调
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                list.stream()
                    .map(message -> new String(message.getBody(), StandardCharsets.UTF_8))
                    .forEach(System.out::println);
                // 返回消费是否成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}
